#include "pubsub.h"
#include "codec.h"

#include <functional>

using namespace placeholders;
using namespace pubsub;

PubSubClient::PubSubClient(EventLoop* loop,
	const InetAddress& hubAddr,
	const string& name) :m_client(loop,hubAddr,name)
{
	m_client.setConnectionCallback(std::bind(&PubSubClient::onConnection, this, _1));
	m_client.setMessageCallback(std::bind(&PubSubClient::onMessage, this, _1, _2, _3));
}

void PubSubClient::start()
{
	m_client.connect();
}

void PubSubClient::stop()
{
	m_client.disconnect();
}

bool PubSubClient::connected()const
{
	return m_ptrConn && m_ptrConn->connected();
}

bool PubSubClient::subscribe(const string& topic, const SubscribeCallback& cb)
{
	string message = "sub" + topic + "\r\n";
	m_subCallback = cb;
	return send(message);
}

void PubSubClient::unsubscribe(const string& topic)
{
	string message = "unsub" + topic + "\r\n";
	send(message);
}

bool PubSubClient::publish(const string& topic, const string& content)
{
	string message = "pub " + topic + "\r\n" + content + "\r\n";
	return send(message);
}

void PubSubClient::onConnection(const TcpConnectionPtr& conn)
{
	if (conn->connected())
	{
		m_ptrConn = conn;
	}
	else
	{
		m_ptrConn.reset();
	}
	if (m_connCallback)
	{
		m_connCallback(this);
	}
}

void PubSubClient::onMessage(const TcpConnectionPtr& conn,
	Buffer* buf,
	Timestamp receiveTime)
{
	ParseResult result = kSuccess;
	while (result == kSuccess)
	{
		string cmd;
		string topic;
		string content;

		result = parseMessage(buf, &cmd, &topic, &content);
		if (result == kSuccess)
		{
			if (cmd == "pub" && m_subCallback)
			{
				m_subCallback(topic, content, receiveTime);
			}
		}
		else if (result == kError)
		{
			conn->shutdown();
		}
	}
}

bool PubSubClient::send(const string& message)
{
	bool succeed = false;
	if (m_ptrConn && m_ptrConn->connected())
	{
		m_ptrConn->send(message);
		succeed = true;
	}
	return succeed;
}